#include "my_worker.h"
#include "my_event_loop.h"

#include "comm/my_hooks.h"
#include "comm/my_random.h"
#include "config/my_cache_conf.h"

typedef struct {
    int                     nthreads;
    my_event_loop_data_t    *datas;
    my_thread_base_t        *threads;
} my_worker_processor_t;

static my_worker_processor_t    worker_processor = {0, NULL, NULL};
static int64_t                  worker_thread_idx = 0;


// run in event loop.
static void
my_worker_event_loop_init(my_event_loop_data_t *loop_data)
{
    my_hook_chain_exec(worker_init_chain, loop_data);
}

static void
my_worker_event_loop_quit(my_event_loop_data_t *loop_data)
{
    my_hook_chain_exec(worker_quit_chain, loop_data);
}

// run in main thread.
void
my_worker_processor_start()
{
    int     nevents;
    int     worker_threads;
    int64_t mem_pool_water_level;

    my_cache_conf_t *cache_conf = my_cache_conf_get();
    my_service_conf_t *service_conf = &cache_conf->service_conf;

    nevents = (int)my_conf_item_int64_get(&service_conf->nevents);
    worker_threads = (int)my_conf_item_int64_get(&service_conf->worker_threads);
    mem_pool_water_level =
        my_conf_item_int64_get(&service_conf->worker_mem_pool_water_level);

    if (nevents <= 0 || worker_threads <= 0) {
        MY_LOG_FATAL("nevents and worker threads must be greate than 0");
    }


    my_event_loop_data_t    *datas;
    my_thread_base_t        *threads;

    datas = my_thread_calloc((int64_t)sizeof(my_event_loop_data_t) * worker_threads);
    threads = my_thread_calloc((int64_t)sizeof(my_thread_base_t) * worker_threads);

    for (int i = 0; i < worker_threads; i++) {
        my_event_loop_data_init(datas+i,
            my_worker_event_loop_init,
            my_worker_event_loop_quit,
            nevents);
        my_thread_base_start(
            threads+i,
            my_event_loop_runner,
            datas+i,
            mem_pool_water_level,
            0);
    }

    worker_processor.nthreads = worker_threads;
    worker_processor.threads = threads;
    worker_processor.datas = datas;
}

// run in main thread.
void
my_worker_processor_stop()
{
    for (int i = 0; i < worker_processor.nthreads; i++) {
        my_event_loop_quit(worker_processor.threads+i);
    }

    for (int i = 0; i < worker_processor.nthreads; i++) {
        my_thread_base_join(worker_processor.threads+i, NULL);
    }

    my_thread_free(worker_processor.datas);
    my_thread_free(worker_processor.threads);
}

void
my_worker_processor_post_event(my_dispatch_event_t *dispatch_event)
{
    if (worker_processor.nthreads == 0) {
        MY_LOG_FATAL("worker_threads must be greater than 0");
    }

    worker_thread_idx = (worker_thread_idx + 1) % worker_processor.nthreads;
    my_event_loop_post_event(worker_processor.threads + worker_thread_idx, dispatch_event);
}